package com.fy.service.server.listener;

import com.alibaba.fastjson.JSONObject;
import com.fy.common.model.im.AttributeChannelModel;
import com.fy.common.model.im.MessageModel;
import com.fy.service.server.handler.SocketServerBusiHandler;
import com.fy.service.server.service.ImService;
import com.fy.service.service.ImBrokerService;
import io.netty.channel.Channel;
import jodd.bean.BeanUtil;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


/**
 * Description：消息下发中心，负责把消息下发到客户端
 */

@Component
public class MessagePushListener implements MessageListenerConcurrently {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    public static ConcurrentHashMap<String, Channel> channelMapping = SocketServerBusiHandler.channelMapping;



    @Autowired
    private ImService imService;
    @Autowired
    private ImBrokerService imBrokerService;
/**
     * 消息下发
     * @param list
     * @param consumeConcurrentlyContext
     * @return
     */

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        try {
            for (MessageExt messageExt : list) {
                byte[] body = messageExt.getBody();
                String msg = new String(body);
                MessageModel model = JSONObject.parseObject(msg,MessageModel.class);
                logger.info("消息到达消息下发中心" + msg);

                //按照协议获取到下发的通道
                Channel channel = channelMapping.get(model.getTenantId() + "_" + model.getTo() + "_" + model.getProtocolType() + "_" + model.getDeviceCode() );
                logger.info("key" + model.getTenantId() + "_" + model.getTo() + "_" + model.getProtocolType() + "_" + model.getDeviceCode());
                if(channel != null){
                    imBrokerService.push(model,channel);
                    logger.info("推送的消息"+ JSONObject.toJSONString(model));
                }
            }
        } catch (Exception e) {
            logger.error("message transpond error " ,e);
        }

        //执行完成消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

    }

}
